package com.beametl.sample.origin;

import java.sql.ResultSet;

import com.beametl.sample.DataRow;
import com.beametl.sample.DataRowUtil;
import com.beametl.sample.entity.StaffVO;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.io.jdbc.JdbcIO;

/*关系型数据库到文本之间的ETL传输样例*/
public class JDBCToJDBCSampleOrigin {

    public static void dealWithPipeline(Pipeline pipeline) {
        JdbcIO.DataSourceConfiguration config1 = JdbcIO.DataSourceConfiguration
                .create("com.ibm.db2.jcc.DB2Driver", "jdbc:db2://10.251.106.56:50400/RBSDBT2").withUsername("xxxxxx")
                .withPassword("xxxxx");
        JdbcIO.DataSourceConfiguration config2 = JdbcIO.DataSourceConfiguration
                .create("com.mysql.jdbc.Driver",
                        "jdbc:mysql://10.110.35.175:3306/biapp?useUnicode=true&characterEncoding=utf-8")
                .withUsername("xxxxx").withPassword("xxxxxx");

        pipeline.apply(JdbcIO.<StaffVO>read().withFetchSize(1000).withDataSourceConfiguration(config1)
                .withQuery("select * from baozhang.wf_staff fetch first 100 rows only")
                .withRowMapper(new JdbcIO.RowMapper<StaffVO>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public StaffVO mapRow(ResultSet rs) throws Exception {

                        StaffVO staffVO = new StaffVO();
                        staffVO.setStaffid(rs.getString("STAFFID"));
                        staffVO.setStaffname(rs.getString("STAFFNAME"));
                        return staffVO;
                    }
                }).withCoder(SerializableCoder.of(StaffVO.class)))

                .apply("writeTargetDB",

                        JdbcIO.<StaffVO>write().withDataSourceConfiguration(config2)
                                .withStatement(String.format("insert into WF_STAFF(STAFFID ,STAFFNAME) values(?,?)"))
                                .withBatchSize(1000).withRetryStrategy((JdbcIO.RetryStrategy) e -> {
                                    return "XJ208".equals(e.getSQLState()); // we
                                                                            // fake
                                                                            // a
                                                                            // deadlock
                                                                            // with
                                                                            // a
                                                                            // lock
                                                                            // here
                                }).withPreparedStatementSetter((element, statement) -> {
                                    // System.out.println((String)element.get("STAFFID")
                                    // +","+ (String)element.get("STAFFNAME"));
                                    statement.setString(1, (String) element.getStaffid());
                                    statement.setString(2, (String) element.getStaffname());
                                }));
    }

}